package com.relytosoft.mizar.analyze.system.services.consumer;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.system.api.entity.AyGroupBase;
import com.relytosoft.mizar.analyze.system.services.service.AyGroupBaseService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * @author lj
 * @date 2022/5/17 9:25
 */
@Component
public class AyGroupBaseConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private AyGroupBaseService ayGroupBaseService;

    @KafkaListener(id = "consumer-mizar-group-sync", topics = MediaMqConstants.FACE_ANALYZE_GROUP_ID_TOPIC, containerFactory = "singleContainerFactory")
    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            boolean isConsume = false;
            String data = kafkaOpMsg.getData();
            if (!StrUtil.isEmptyIfStr(data)) {
                Map<String, Object> map = JSONObject.parseObject(data, new TypeReference<Map<String, Object>>() {
                });
                //我们库的库ID
                Integer id = (Integer) map.get("id");
                //三方库的库ID
                String groupId = (String) map.get("groupId");
                //根据库id查询自己库信息并更新三方库ID到数据库
                AyGroupBase ayGroupBase = ayGroupBaseService.getById(id);
                ayGroupBase.setPlateformGroupId(groupId);
                isConsume = ayGroupBaseService.updateById(ayGroupBase);
            }
            log.info("线程:{} 消费->{}", Thread.currentThread(), kafkaOpMsg.getOpType());
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());
        }
    }
}
